package cn.mengfly.testnrz.reactor;

import cn.mengfly.testnrz.nio.NioDemoConfig;
import cn.mengfly.testnrz.util.IOUtil;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 多线程的反应器模式
 *
 * @author Mengfly
 */
public class TestMultiThreadReactor {

    static class MultiThreadEchoServerReactor {
        ServerSocketChannel serverSocket;
        AtomicInteger next = new AtomicInteger(0);
        // 引入多个选择器
        List<SubReactor> subReactors;

        MultiThreadEchoServerReactor() throws IOException {
            subReactors = new ArrayList<>();
            serverSocket = ServerSocketChannel.open();
            InetSocketAddress address = new InetSocketAddress(NioDemoConfig.SOCKET_PORT);
            serverSocket.socket().bind(address);
            serverSocket.configureBlocking(false);

            // 添加子反应器1
            addReactor(new SubReactor());
            // 添加子反应器2
            addReactor(new SubReactor());

            // 第一个选择器负责监控新的连接事件
            attachHandler(serverSocket, new AcceptorHandler(serverSocket, this), SelectionKey.OP_ACCEPT);
        }

        /**
         * 添加反应器
         */
        public void addReactor(SubReactor reactor) {
            subReactors.add(reactor);
        }

        /**
         * 为通道绑定事件类型
         * @param channel 通道
         * @param handler 处理器
         * @param key 要绑定的IO事件
         */
        public void attachHandler(SelectableChannel channel, Object handler, int key) throws IOException {
            SelectionKey sk = subReactors.get(next.get()).register(channel, key);
            sk.attach(handler);
            if (next.incrementAndGet() >= subReactors.size()) {
                next.set(0);
            }
        }

        /**
         * 启动服务
         */
        public void startService() {
            for (SubReactor subReactor : subReactors) {
                new Thread(subReactor).start();
            }
        }
    }

    static class MultiThreadEchoHandler implements Runnable {
        final SocketChannel channel;
        final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        // 线程池
        static ExecutorService pool = Executors.newFixedThreadPool(4);

        MultiThreadEchoHandler(SocketChannel socketChannel) {
            this.channel = socketChannel;
        }

        @Override
        public void run() {
            pool.execute(new AsyncTask());
        }


        private class AsyncTask implements Runnable {
            @Override
            public void run() {
                MultiThreadEchoHandler.this.asyncRun();
            }
        }

        private void asyncRun() {
            int length;
            try {
                while ((length = channel.read(byteBuffer)) > 0) {
                    System.out.println(new String(byteBuffer.array(), 0, length));
                    byteBuffer.clear();
                }
            } catch (IOException e) {
                e.printStackTrace();
                IOUtil.close(channel);
            }
        }
    }

    static class AcceptorHandler implements Runnable {
        private final MultiThreadEchoServerReactor reactorManager;
        private final ServerSocketChannel serverSocket;

        public AcceptorHandler(ServerSocketChannel serverSocket, MultiThreadEchoServerReactor reactorManager) {
            this.reactorManager = reactorManager;
            this.serverSocket = serverSocket;
        }

        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocket.accept();
                if (channel != null) {
                    channel.configureBlocking(false);
                    reactorManager.attachHandler(channel, new MultiThreadEchoHandler(channel), SelectionKey.OP_READ);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 子反应器
     */
    static class SubReactor implements Runnable {
        final Selector selector;

        public SubReactor() throws IOException {
            selector = Selector.open();
        }

        /**
         * 为自选择器注册要轮询的IO事件
         * @param channel 通道
         * @param op IO事件类型
         */
        public SelectionKey register(SelectableChannel channel, int op) throws IOException {
            SelectionKey register = channel.register(selector, op);
            selector.wakeup();
            return register;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    for (SelectionKey sk : selectionKeys) {
                        dispatch(sk);
                    }
                    selectionKeys.clear();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private void dispatch(SelectionKey sk) {
            Runnable handler = (Runnable) sk.attachment();
            if (handler != null) {
                handler.run();
            }
        }
    }

    public static void main(String[] args) throws IOException {
        MultiThreadEchoServerReactor server = new MultiThreadEchoServerReactor();
        server.startService();
    }

}
